package com.yz.order.mq;

import com.rabbitmq.client.Channel;
import com.yz.common.constant.MQConstants;
import com.yz.model.entity.order.Order;
import com.yz.order.service.OrderService;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;

/**
 * Created With IntelliJ IDEA.
 * Descriptions:
 * <p>
 * User:Mr.Du
 * Date:2024/4/9
 * Time:13:43
 */
@Slf4j
@Component
public class OrderMessageConsumer {

    @Resource
    private OrderService orderService;

    private Integer MAX_RETRY_COUNT = 3;

    private Integer cnt = 0;

    /**
     * 监听MQ中的订单队列，接收消息并处理。
     * 使用手动确认模式（ACK），需要手动确认消息处理成功。
     *
     * @param message 接收到的消息内容
     * @param channel AMQP通道，用于与消息队列进行交互
     * @param deliveryTag 消息的投递标签，用于确认或拒绝消息
     * @throws Exception 如果消息处理过程中发生异常，则抛出异常
     */
    @SneakyThrows
    @RabbitListener(queues = {MQConstants.ORDER_QUEUE_NAME}, ackMode = "MANUAL")
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try{
            // 处理接收到的消息
            this.handleMessage(message, channel, deliveryTag);
        }catch (Exception e){
            // 记录消息处理失败的日志，并进行否定确认（NACK），同时重抛异常
            log.error("消息消费失败order", e.getMessage());
            if(++cnt > MAX_RETRY_COUNT){
                channel.basicNack(deliveryTag, false, false);
            }else{
                channel.basicNack(deliveryTag, false, true);
            }
        }
    }

    /**
     * 处理消息的函数。
     * 对接收到的消息进行处理，首先检查消息是否为空，若为空则拒绝消息且抛出异常。
     * 非空消息会被解析为订单ID，并尝试更新订单状态。如果更新失败，则拒绝消息并抛出异常。
     * 成功处理消息后会确认消息消费。
     *
     * @param message 接收到的消息内容，预期为订单ID的字符串形式。
     * @param channel AMQP消息通道，用于确认或拒绝消息。
     * @param deliveryTag 消息的唯一标识，用于在确认或拒绝消息时使用。
     * @throws IOException 如果发生IO错误。
     * @throws RuntimeException 如果消息为空或更新订单状态失败。
     */
    @Transactional
    void handleMessage(String message, Channel channel, long deliveryTag) throws IOException {
        log.info("receiveMessage message = {}", message);
        if(StringUtils.isBlank(message)){
            channel.basicNack(deliveryTag, false, false); // 拒绝消息，不重试
            throw new RuntimeException("消息为空");
        }
        Long id = Long.parseLong(message); // 将消息解析为订单ID
        Order order = Order.builder().build(); // 初始化订单
        order.setStatus(1); // 设置订单状态为已处理
        order.setId(id); // 设置订单ID
        boolean update = orderService.updateById(order); // 根据ID更新订单状态
        if(!update){
            channel.basicNack(deliveryTag, false, false); // 更新失败，拒绝消息，不重试
            throw new RuntimeException("更新订单状态失败");
        }
        channel.basicAck(deliveryTag, false); // 成功处理消息，确认消费
    }

}